package utilities;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import dataFromXml.Join;
import utilDataStruct.FileStruct;

public class JoinHandler {
	public static void JoinMRFile(String tFolder, String frameFolder, 
			List<FileStruct> fileList, String joinType) {

		// judge whether the template is here.
		String filePath = tFolder + File.separator + "MR_JoinTables.java";
		File tarFile = new File(filePath);
		if (tarFile.exists()) {
			return;
		}

		// if there is no target file, create it.
		BufferedWriter bw, bw_import;
		BufferedReader br;
		try {
			bw = new BufferedWriter(new FileWriter(filePath));
			bw_import = new BufferedWriter(new FileWriter(tFolder + File.separator + "tempImport", true));
			br = new BufferedReader(new FileReader(frameFolder + "TableJoinT"));

			String line;
			String[] strArray;

			while ((line = br.readLine()) != null) {
				strArray = line.split("#");
				if (strArray.length > 1) {
					if (strArray[1].equals("0")) {
						String[] pathFolder = tFolder.split("/");
						strArray[1] = pathFolder[pathFolder.length - 1];
					} else if (strArray[1].equals("1")) {
						strArray[1] = joinType;
					} else if (strArray[1].equals("2")) {
						strArray[1] = getTableList(fileList);
					} else if (strArray[1].equals("3")) {
						strArray[1] = getAttrNumList(fileList);
					}
					line = strArray[0] + strArray[1] + strArray[2];
				}
				bw.write(line + "\n");
				bw.flush();
			} // while
			bw.write("\n");
			bw.flush();

			strArray = tFolder.split("/");
			bw_import.write("import " + strArray[strArray.length-1] + ".MR_JoinTables.JoinMapperClass;\n");
			bw_import.write("import " + strArray[strArray.length-1] + ".MR_JoinTables.JoinReducerClass;\n");
			bw_import.write("import " + strArray[strArray.length-1] + ".MR_JoinTables.TaggedRecordWritable;\n");
			bw_import.write("import org.apache.hadoop.mapred.JobClient;\n");
			bw_import.write("import org.apache.hadoop.mapred.JobConf;\n");
			bw_import.flush();
			
			bw.close();
			br.close();
			bw_import.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			bw = null;
			br = null;
		}
	}

	public static String JoinProfile(String tFolder, String frameFolder, List<FileStruct> fileList,
			List<Integer> fnList, Join joinNode, int jobNum) {
		
		String attributesList = null;
		BufferedWriter bw;
		BufferedReader br;
		try {
			bw = new BufferedWriter(new FileWriter(tFolder + File.separator + "tempProfile", true));
			br = new BufferedReader(new FileReader(frameFolder + "TableJoinP"));
			
			String line;

			while ((line = br.readLine()) != null) {
				
				String[] strArray = line.split("#");
				if (strArray.length > 1) {
					if (strArray[1].equals("2")) {
						attributesList = getAttributesList(fileList, joinNode);
						strArray[1] = attributesList;
					} else if (strArray[1].equals("0")) {
						strArray[1] = jobNum + "";
					
						if (strArray.length > 3) {
							strArray[3] = getMultiPaths(fileList, fnList);
						}
					} else {
						assert false : "Table join profile may be error!";
					}
				}
				line = "";
				for (int i = 0; i < strArray.length; i++) {
					line += strArray[i];
				}
				bw.write(line + "\n");
				bw.flush();
			}
			bw.write("\n");
			bw.flush();

			bw.close();
			br.close();
			System.out.println("Table join profile is OK!");
		} catch (IOException ex) {
			ex.printStackTrace();
		}
		return attributesList;
	}
	
	private static String getMultiPaths(List<FileStruct> fileList, List<Integer> fnList) {

		StringBuilder multiPaths = new StringBuilder();
		
		multiPaths.append("new Path(args[1] + \"/temp%_%" + fileList.get(0).getTableName());
		multiPaths.append("%_%" + fnList.get(0) + "\")");
		
		for (int i=1; i<fileList.size(); i++) {
			multiPaths.append(", new Path(args[1] + \"/temp%_%");
			multiPaths.append(fileList.get(i).getTableName());
			multiPaths.append("%_%" + fnList.get(i) + "\")");
		}
		return multiPaths.toString();
	}

	private static String getAttributesList(List<FileStruct> fileList, Join joinNode) {
		
		StringBuilder attributes = new StringBuilder();
		
		String[] tnArray = joinNode.getTableNames().split(" ");
		String[] jaArray = joinNode.getJoinAttr().split(" ");
		
		List<String> tableNames = new ArrayList<>();
		for (FileStruct fs : fileList) {
			tableNames.add(fs.getTableName());
		}
		Collections.sort(tableNames);
		for (int i=0; i<tableNames.size(); i++) {
			int j;
			for (j = 0; j<tnArray.length; j++) {
				if (tnArray[j].equals(tableNames.get(i))) {
					break;
				}
			}
			if (i == 0) {
				attributes.append(jaArray[j]);
			}
			
			for (FileStruct fs : fileList) {
				if (tnArray[j].equals(fs.getTableName())) {
					for (int k=0; k<fs.getAttrNameList().size(); k++) {
						if (!jaArray[j].equals(fs.getAttrNameList().get(k))) {
							attributes.append(" " + fs.getAttrNameList().get(k));
						}
					}
				}
			}
		}
		return attributes.toString();
	}
	
	private static String getAttrNumList(List<FileStruct> fileList) {
		
		StringBuilder attrNumList = new StringBuilder();
		List<String> attrList;
		List<String> tableNames = new ArrayList<>();
		for (FileStruct fs : fileList) {
			tableNames.add(fs.getTableName());
		}
		Collections.sort(tableNames);
		
		for (int i=0; i<tableNames.size(); i++) {
			if (i != 0)
				attrNumList.append(" ");
			for (int j=0; j<fileList.size(); j++) {
				if (tableNames.get(i).equals(fileList.get(j).getTableName())) {
					attrList = fileList.get(j).getAttrNameList();
					attrNumList.append(attrList.size() - 1);
					break;
				}
			}
		}
		return attrNumList.toString();
	}
	
	private static String getTableList(List<FileStruct> fileList) {

		StringBuilder tableList = new StringBuilder();
		List<String> tableNames = new ArrayList<>();
		for (FileStruct fs : fileList) {
			tableNames.add(fs.getTableName());
		}
		Collections.sort(tableNames);
		
		for (int i=0; i<tableNames.size(); i++) {
			if (i != 0)
				tableList.append(" ");
			tableList.append(tableNames.get(i));
		}
		return tableList.toString();
	}
}
